package cn.azzhu.day02

import org.apache.flink.streaming.api.functions.co.{CoFlatMapFunction, CoMapFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * @author azzhu
 * @create 2020-09-21 00:17:24
 */
object CoFlatMapExample {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream1 = env.fromElements(
      ("zs", 130),
      ("lisi", 150)
    )

    val stream2 = env.fromElements(
      ("zs", 35),
      ("lisi", 23)
    )

    //直接connect两条流没有任何意义
    //必须把相同key的流联合在一起处理
    val connected = stream1.keyBy(_._1)
      .connect(stream2.keyBy(_._1))

    val printed = connected.flatMap(new MyCoFlatMapFunction)
    printed.print()

    env.execute("CoFlatMapExample")
  }

  class MyCoFlatMapFunction extends CoFlatMapFunction[(String,Int),(String,Int),String] {
    override def flatMap1(value: (String, Int), out: Collector[String]): Unit = {
      out.collect(value._1 + "的体重是:" + value._2 + "斤")
      out.collect(value._1 + "的体重是:" + value._2 + "斤")
    }

    override def flatMap2(value: (String, Int), out: Collector[String]): Unit = {
      out.collect(value._1 + "的年龄是:" + value._2 + "岁")
    }
  }

}
